package com.dataloader;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.serializer.SerializerFeature;
import com.bean.*;
import com.utils.RedisUtil;
import redis.clients.jedis.Jedis;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.ResultSet;
import java.sql.Statement;

/**
 * @Description: 装载全量数据
 * @Author: Sky
 * @Times : 2021/8/11 16:10
 * 维度表数据的全量装载实现类
 *   * 需要将以下五个维度表的数据同步到redis中
 *   * 1）商品维度表
 *   * 2）商品分类维度表
 *   * 3）店铺表
 *   * 4）组织机构表
 *   * 5）门店商品分类表
 */
public class DimentsionDataLoader {

  public static void main(String[] args) throws Exception {

      Class.forName("com.mysql.cj.jdbc.Driver");
      Connection connection = DriverManager.getConnection("jdbc:mysql://hadoop106:3306/FlinkEtl?useSSL=false&serverTimezone=UTC&characterEncoding" +
          "=UTF-8", "root", "root");
      //创建Redis连接
      Jedis jeids = RedisUtil.getJedis().getResource();
      //保存数据到第一个数据路
      jeids.select(1);
       LoadDimGoods(connection,jeids);
       loadDimShops(connection,jeids);
      loadDimGoodsCats(connection,jeids);
      loadDimOrg(connection,jeids);
      LoadDimShopCats(connection,jeids);
    connection.close();

  }


  //加载商品维度表数据到Redis
  public static void LoadDimGoods(Connection connection,Jedis jedis) throws Exception{
    //定义SQL语句
    String sql="select goodsId,goodsName,shopId,goodsCatId,shopPrice from foo_goods";
    Statement statement = connection.createStatement();
    ResultSet resultSet = statement.executeQuery(sql);
    while(resultSet.next()){
      Long goodsId = resultSet.getLong("goodsId");
      String goodsName = resultSet.getString("goodsName");
      Long shopId = resultSet.getLong("shopId");
      Integer goodsCatId = resultSet.getInt("goodsCatId");
      Double shopPrice = resultSet.getDouble("shopPrice");
      DimGoodsDBEntity goodsDBEntity=new DimGoodsDBEntity(goodsId,goodsName,shopId,goodsCatId,shopPrice);
      System.out.println(goodsDBEntity);
      //将对象转换为JSON字符串
      String  goodDB= JSON.toJSONString(goodsDBEntity, SerializerFeature.DisableCircularReferenceDetect);
      jedis.hset("foo_shop:dim_goods",goodsId.toString(),goodDB);
    }
    resultSet.close();
//    statement.close();
  }


  //加载商铺维度数据到Redis
  public static void  loadDimShops(Connection connection,Jedis jedis) throws Exception {
    //定义SQL语句
    String sql = "select t1.shopId,t1.areaId,t1.shopName,t1.shopCompany from foo_shops t1";
    Statement statement = connection.createStatement();
    ResultSet resultSet = statement.executeQuery(sql);
    while (resultSet.next()) {
      Integer shopId = resultSet.getInt("shopId");
      Integer areaId = resultSet.getInt("areaId");
      String shopName = resultSet.getString("shopName");
      String shopCompany = resultSet.getString("shopCompany");
      DimShopsDBEntity goodsDBEntity = new DimShopsDBEntity(shopId, areaId, shopName, shopCompany);
      System.out.println(goodsDBEntity);
      //将对象转换为JSON字符串
      String goodDB = JSON.toJSONString(goodsDBEntity, SerializerFeature.DisableCircularReferenceDetect);
      jedis.hset("foo_shop:dim_shops", shopId.toString(), goodDB);

    }
      resultSet.close();
//      statement.close();
  }


  // 加载商品分类维度数据到Redis
  public static void loadDimGoodsCats(Connection connection,Jedis jedis) throws Exception{
    String sql="select t1.catId,t1.parentId,t1.catName,t1.cat_level from foo_goods_cats t1";
    Statement statement = connection.createStatement();
    ResultSet resultSet = statement.executeQuery(sql);
    while(resultSet.next()){
      String catId = resultSet.getString("catId");
      String parentId = resultSet.getString("parentId");
      String catName = resultSet.getString("catName");
      String cat_level = resultSet.getString("cat_level");
      DimGoodsCatDBEntity dimGoodsCatDBEntity=new DimGoodsCatDBEntity(catId,parentId,catName,cat_level);
      System.out.println(dimGoodsCatDBEntity);
      String goodDB = JSON.toJSONString(dimGoodsCatDBEntity, SerializerFeature.DisableCircularReferenceDetect);
      jedis.hset("foo_shop:dim_goods_cats", catId, goodDB);
    }
    resultSet.close();

  }


  //加载组织结构维度数据
  public static void loadDimOrg(Connection connection,Jedis jedis) throws Exception{
    String sql="select orgid,parentid,orgName,orgLevel from foo_org";
    Statement statement = connection.createStatement();
    ResultSet resultSet = statement.executeQuery(sql);
    while(resultSet.next()){
      Integer orgId = resultSet.getInt("orgId");
      Integer parentId = resultSet.getInt("parentId");
      String orgName = resultSet.getString("orgName");
      Integer orgLevel = resultSet.getInt("orgLevel");
      DimOrgDBEntity dimOrgDBEntity=new DimOrgDBEntity(orgId,parentId,orgName,orgLevel);
      System.out.println(dimOrgDBEntity);
      String goodDB = JSON.toJSONString(dimOrgDBEntity, SerializerFeature.DisableCircularReferenceDetect);
      jedis.hset("foo_shop:dim_org", orgId+"", goodDB);
    }
    resultSet.close();
//    connection.close();
  }


  //加载门店商品分类维度数据到Redis
  public static void LoadDimShopCats(Connection connection,Jedis jedis) throws Exception{
     String sql="select t1.catId,t1.parentId,t1.catName,t1.catSort from foo_shop_cats t1";
    Statement statement = connection.createStatement();
    ResultSet resultSet = statement.executeQuery(sql);
    while(resultSet.next()){
      String catId = resultSet.getString("catId");
      String parentId = resultSet.getString("parentId");
      String catName = resultSet.getString("catName");
      String cat_level = resultSet.getString("catSort");
      DimShopCatDBEntity dimShopCatDBEntity=new DimShopCatDBEntity(catId,parentId,catName,cat_level);
      //System.out.println(dimShopCatDBEntity);
      String goodDB = JSON.toJSONString(dimShopCatDBEntity, SerializerFeature.DisableCircularReferenceDetect);
      jedis.hset("foo_shop:dim_shop_cats", catId, goodDB);
      String res = jedis.hget("foo_shop:dim_shop_cats", catId);
      System.out.println(res);


    }
    resultSet.close();
//    connection.close();
  }


}
